package org.jeecg.modules.iot.aio.task;

import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.iot.model.v20180120.PubRequest;
import com.aliyuncs.iot.model.v20180120.PubResponse;
import org.jeecg.modules.iot.aio.ChannelContext;
import org.jeecg.modules.iot.aio.message.CheckMessage;
import org.jeecg.modules.iot.aio.task.base.AbstractQueueRunnable;
import org.jeecg.modules.iot.util.IotClientUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.TcpConst;
import org.tio.utils.queue.FullWaitQueue;
import org.tio.utils.queue.TioFullWaitQueue;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import static org.apache.commons.codec.binary.Base64.encodeBase64String;

/**
 * 这里是一个注释
 *
 * @author scw
 * @version v1.0
 * @date
 */
public class CheckMessageRunnable extends AbstractQueueRunnable<CheckMessage> {

    private static final Logger log	 = LoggerFactory.getLogger(CheckMessageRunnable.class);

    /** The msg queue. */
    private FullWaitQueue<CheckMessage> msgQueue = null;
    public boolean canSend = true;

    public ChannelContext channelContext;

    private static final int	MAX_CAPACITY_MIN	= TcpConst.MAX_DATA_LENGTH - 1024;	//减掉1024是尽量防止溢出的一小部分还分成一个tcp包发出
    private static final int	MAX_CAPACITY_MAX	= MAX_CAPACITY_MIN * 10;

    public CheckMessageRunnable(ThreadPoolExecutor executor, ChannelContext channelContext) {
        super(executor);
        this.channelContext = channelContext;
    }

    @Override
    public void runTask() {
        if (msgQueue.isEmpty())return;
        IotClientUtil iotClientUtil = this.channelContext.aioConfiguration.getIotClientUtil();
        int queueSize = msgQueue.size();
        if(queueSize == 1){
            CheckMessage message = msgQueue.poll();
            if(message != null){
                checkMessage(message,iotClientUtil);
            }
        }
        List<CheckMessage> messageList = new ArrayList<>();
        CheckMessage message = null;
        boolean sslChanged = false;
        while ((message = msgQueue.poll()) != null) {
            if(canSend || sslChanged)break;
            messageList.add(message);
        }
        checkMessageList(messageList,iotClientUtil);
    }

    public void checkMessageList(List<CheckMessage> messageList,IotClientUtil iotClientUtil){
        for(CheckMessage message:messageList){
            checkMessage(message,iotClientUtil);
        }
    }


    public void checkMessage(CheckMessage message,IotClientUtil iotClientUtil){
        String topic = "/" + message.getProductKey() + "/" + message.getSn()  + "/user/get";
        try {
            byte[] bytes = message.getMessage().getBytes(StandardCharsets.UTF_8);
            PubResponse response = sendMsgAsync(message.getProductKey(), topic, bytes, 1);
            this.channelContext.isCheck = true;
            this.channelContext.needQuery = true;
            this.channelContext.response = response;
            /**
             * 查询超时时间15秒
             */
            this.channelContext.checkExpireTime = System.currentTimeMillis()+1500;
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 发送异步消息
     */
    public PubResponse sendMsgAsync(String productKey, String topicFullName, byte[] bytes, int qos) throws ClientException {
        PubRequest request = new PubRequest();
        String base64 = encodeBase64String(bytes);
        request.setProductKey(productKey);
        request.setMessageContent(base64);
        request.setTopicFullName(topicFullName);
        IotClientUtil iotClientUtil = channelContext.aioConfiguration.getIotClientUtil();
        request.setIotInstanceId(iotClientUtil.getIotInstanceId());
        request.setQos(qos > 1 ? 1 : qos); //目前支持QoS0和QoS1
        PubResponse response = channelContext.aioConfiguration.getIotClientUtil().getDefaultAcsClient().getAcsResponse(request);
        return response;
    }

    @Override
    public FullWaitQueue<CheckMessage> getMsgQueue() {
        if (msgQueue == null) {
            synchronized (this) {
                if (msgQueue == null) {
                    msgQueue = new TioFullWaitQueue<CheckMessage>(Integer.getInteger("tio.fullqueue.capacity", null), false);
                }
            }
        }
        return msgQueue;
    }

}
